package org.budo.dubbo.registry.redis;

import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.budo.dubbo.registry.redis.util.BudoRedisRegistryRepository;
import org.budo.dubbo.registry.redis.util.DubboUrlUtil;
import org.budo.dubbo.registry.redis.util.RegistryJedisUtil;

import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.common.utils.CollectionUtils;
import com.alibaba.dubbo.common.utils.StringUtils;
import com.alibaba.dubbo.registry.NotifyListener;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

/**
 * @author li
 * @version 1 (2015年11月14日 下午1:56:55)
 * @since Java7
 * @see com.alibaba.dubbo.registry.redis.RedisRegistry.NotifySub
 */
public class BudoRedisRegistryPubSub extends JedisPubSub {
    private static final Logger log = LoggerFactory.getLogger(BudoRedisRegistryPubSub.class);

    private static final Map<URL, NotifyListener> notifyListenerMap = new ConcurrentHashMap<URL, NotifyListener>();

    private BudoRedisRegistryRepository redisRegistryRepository = new BudoRedisRegistryRepository();

    private AbstractBudoRedisRegistry budoRedisRegistry;

    public BudoRedisRegistryPubSub(AbstractBudoRedisRegistry budoRedisRegistry) {
        this.budoRedisRegistry = budoRedisRegistry;
    }

    /**
     * @see com.alibaba.dubbo.registry.redis.RedisRegistry#doNotify(Jedis, Collection<String>, URL, Collection<NotifyListener>)
     */
    public void onPMessage(String pattern, String key, String message) {
        if (StringUtils.isEquals(Constants.REGISTER, message) || StringUtils.isEquals(Constants.UNREGISTER, message)) { // 有效的消息
            this.processMessage(pattern, key, message);
        } else {
            log.info("#50 onPMessage, pattern=" + pattern + ", key=" + key + ", message=" + message);
        }
    }

    /**
     * 处理收到的推送消息
     */
    public void processMessage(String pattern, String key, String message) {
        Jedis jedis = null;
        try {
            jedis = RegistryJedisUtil.getJedisResource(budoRedisRegistry); // 如果获取链接异常, 这个消息就未通知到
            this.doProcessMessage(jedis, pattern, key, message); // 共用连接 
        } catch (Throwable e) {
            log.error("#63 doProcessMessage error, pattern=" + pattern + ", key=" + key + ", message=" + message + ", e=" + e, e);
        } finally {
            RegistryJedisUtil.returnResource(budoRedisRegistry, jedis);
        }
    }

    /**
     * doProcessMessage
     */
    public void doProcessMessage(Jedis jedis, String pattern, String key, String message) {
        Map<URL, NotifyListener> notifyListenerMap = this.getNotifyListenerMap();
        Set<Entry<URL, NotifyListener>> entrySet = notifyListenerMap.entrySet();
        int size = notifyListenerMap.size();
        int index = 0;

        for (Entry<URL, NotifyListener> entry : entrySet) {
            index++;

            URL notifyListenerUrl = entry.getKey();
            NotifyListener notifyListener = entry.getValue();

            // 如果这里jedis查询异常, 则跳出for循环 
            String notifyKey = key.replace(AbstractBudoRedisRegistry.PUBLISH_CHANNEL_PREFIX, "");
            List<URL> notifyUrls = redisRegistryRepository.listMatchUrlsForProcessMessage(jedis, notifyKey, notifyListenerUrl);

            if (!CollectionUtils.isEmpty(notifyUrls)) { // 非空才通知
                String notifyListenerUrlStr = DubboUrlUtil.toSimpleString(notifyListenerUrl);

                log.info("processMessage, " + index + "/" + size + ", pattern=" + pattern + ", key=" + key + ", message=" + message + ", notifyListenerUrl=" // 
                        + notifyListenerUrlStr + ", notifyListener=" + notifyListener + ", notifyUrls=" + DubboUrlUtil.toSimpleString(notifyUrls));
                notifyListener.notify(notifyUrls);
            }
        }
    }

    public Map<URL, NotifyListener> getNotifyListenerMap() {
        return notifyListenerMap;
    }
}